package david.java.flink_sql.streamApiIntegration;


import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.Instant;

/**
 * @Description:
 * @See https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/table/data_stream_api/#examples-for-todatastream
 * @Author: ZhaoDawei
 * @Date: Create in 10:14 上午 2022/3/29
 */
public class S3_3_toDataStream {


    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql(
                "CREATE TABLE GeneratedTable "
                        + "("
                        + "  name STRING,"
                        + "  score INT,"
                        + "  event_time TIMESTAMP_LTZ(3),"
                        + "  WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND"
                        + ")"
                        + "WITH ('connector'='datagen')");

        Table table = tableEnv.from("GeneratedTable");


        // === EXAMPLE 1 ===

        // use the default conversion to instances of Row

        // since `event_time` is a single rowtime attribute, it is inserted into the DataStream
        // metadata and watermarks are propagated
        DataStream<Row> dataStream = tableEnv.toDataStream(table);


        // === EXAMPLE 2 ===

        // a data type is extracted from class `User`,
        // the planner reorders fields and inserts implicit casts where possible to convert internal
        // data structures to the desired structured type

        // since `event_time` is a single rowtime attribute, it is inserted into the DataStream
        // metadata and watermarks are propagated
        DataStream<User> dataStream1 = tableEnv.toDataStream(table, User.class);


        // data types can be extracted reflectively as above or explicitly defined
        tableEnv.toDataStream(
                table,
                DataTypes.STRUCTURED(
                        User.class,
                        DataTypes.FIELD("name", DataTypes.STRING()),
                        DataTypes.FIELD("score", DataTypes.INT()),
                        DataTypes.FIELD("event_time", DataTypes.TIMESTAMP_LTZ())
                )
        );

    }


    // POJO with mutable fields
    // since no fully assigning constructor is defined, the field order
    // is alphabetical [event_time, name, score]
    public static class User {

        public String name;

        public Integer score;

        public Instant event_time;
    }
}
